package com.alinesno.cloud.base.logger.consumer;

import java.util.Date;
import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.alinesno.cloud.base.logger.entity.LogMethodRuntimeEntity;
import com.alinesno.cloud.base.logger.service.ILogMethodRuntimeService;
import com.alinesno.cloud.monitor.agent.aspect.BaseAspect;

/**
 * 异常消息监听
 * 
 * @author WeiXiaoJin
 * @since 2020年1月25日 下午19:45:08
 */
@Component
class MontorMethodLoggerConsumerListener extends BaseConsumer {

	private static final Logger log = LoggerFactory.getLogger(MontorMethodLoggerConsumerListener.class);

	@Autowired
	private ILogMethodRuntimeService logMethodRuntimeService;

	/**
	 * 方法消息
	 * 
	 * @param record
	 * @param data
	 * @param partition
	 * @param topic
	 * @param ts
	 */
	@KafkaListener(topics = { BaseAspect.WATCHER_TOPIC_METHOD_RUNTIME })
	public void listenLogger(ConsumerRecord<?, ?> record, @Payload String data,
			@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
			@Header(KafkaHeaders.RECEIVED_TOPIC) String topic, @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts) {

		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {

			Object message = kafkaMessage.get();
			log.debug("---------------->>>>  message = topic：" + topic + ", " + message);

			if (message != null) {
				JSONObject json = JSONObject.parseObject(message + "");
				
				String methodKey = json.getString("methodKey") ; // 方法唯一值
				LogMethodRuntimeEntity e = logMethodRuntimeService.findByMethodKey(methodKey) ; 
			
				if(e == null) {
					String applicationName = json.getString(APPLICATIONNAME);
					String ip = json.getString(IP);
					String hostname = json.getString(HOSTNAME);
					
					// 通过【类包+包名+方法名+参数】为唯一码
					String className = json.getString("className");
					String methodName = json.getString("methodName");
					String methodParams = json.getString("methodParams");
					long createTime = json.getLongValue("createTime");
					long methodTime = json.getLongValue("methodTime");
				
					// 设置值
					e = new LogMethodRuntimeEntity() ; 
					e.setApplicationName(applicationName);
					e.setIp(ip);
					e.setHostname(hostname);
					e.setCreateTime(new Date(createTime));
					e.setMethodTime(methodTime);
					e.setMethod(methodName);
					e.setMethodDesc(className);
					e.setParams(methodParams);
					e.setAddTime(new Date(createTime));
				}
				
				String type = json.getString("type"); // 类型(接口/实类/数据层)(Type)
				int rps = json.getIntValue("rps") ; // 每秒请求数(RPS)
//				int count = json.getIntValue("count") ; // 总请求数(Count)
				double avg = json.getDoubleValue("avg") ; // 平均响应时间(Avg)
				double min = json.getDoubleValue("min") ; // 最小响应时间(Min)
				double max = json.getDoubleValue("max") ; // 最大响应时间(Max)
				int fail = json.getIntValue("fail") ; // 异常次数(Fail)
				int ok = json.getIntValue("ok") ; // 成功次数(OK)
			
//				private String type; // 类型(接口/实类/数据层)(Type)
//				private String timePercent; // 每分钟调用次数(TimePercent)
//				private int rps; // 每秒请求数(RPS)
//				private int count; // 总请求数(Count)
//				private double totalrunTime ; // 总的请求时间 
//				private double avg; // 平均响应时间(Avg)
//				private double min; // 最小响应时间(Min)
//				private double max; // 最大响应时间(Max)
//				private int fail; // 异常次数(Fail)
//				private int ok; // 成功次数(OK)

				e.setMethodKey(methodKey);
				e.setType(type);
				e.setRps(rps);
				e.setCount(e.getCount()+1);
				e.setAvg(avg);
				e.setMin(min);
				e.setMax(max);
				e.setFail(e.getFail()+fail);
				e.setOk(e.getOk()+ok);
				
				logMethodRuntimeService.save(e);
			}

		}
	}

}
